package com.star.batchhandler.demo;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import javax.sql.DataSource;
/**
 * 
 * @author yaohw
 *
 */
public class SimpleSQLCommandScheduler implements ISQLCommandScheduler {
	public volatile Integer jobs = 0;
	private Object sycLock = new Object();
	public final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(SQLBatchOperation.jobsInBatch);
	private Connection conn;
	Statement stmt;
	private volatile int currentJobs;
	private int[] batchResults;
	private ArrayList<Command> commands = new ArrayList<Command>();
	private ArrayList<Command> backup_commands = new ArrayList<Command>();
	private DataSource ds;
	
	public SimpleSQLCommandScheduler(DataSource ds){
		this.ds = ds;
	}

	@Override
	public void run() {
		jobs = 0;
			try {
				conn = ds.getConnection();
				conn.setAutoCommit(false);
			} catch (SQLException e) {
				throw new IllegalStateException(e);
			}
		for (int j = 0; j < SQLBatchOperation.jobsInBatch; j++) {
//			new Thread(new BatchJob(this,ds)).start();
			scheduler.execute(new BatchJob(this,ds));
		}
		while(true){
			if(getJobs().equals(0)){
				break;
			}
			synchronized (sycLock) {
				try {
					sycLock.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public void addCommand(Command cmd) throws Exception {
		synchronized (this) {
			if (cmd.getParams() == null) {
				if (stmt == null) {
					stmt = conn.createStatement();
				}
				stmt.addBatch(cmd.getSql());
			} else {
				if (stmt == null) {
					stmt = conn.prepareStatement(cmd.getSql());
				}
				((PreparedStatement) stmt).setLong(1, (Long) cmd.getParams()[0]);
				((PreparedStatement) stmt).addBatch();
			}
			commands.add(cmd);
			currentJobs++;
			if (currentJobs == SQLBatchOperation.jobsInBatch) {
				batch();
			} else {
				this.wait();
			}
		}
	}

	private void batch() throws Exception {
		batchResults = stmt.executeBatch();
		conn.commit();
		stmt.close();
		recover();
		synchronized (this) {
			this.notifyAll();
		}
	}

	private void recover() {
		backup_commands.clear();
		backup_commands.addAll(commands);
		commands.clear();
		currentJobs = 0;
		stmt = null;
	}

	public int getBatchResult(Command cmd) {
		return /*batchResults[backup_commands.indexOf(cmd)]*/1;
	}

	public Integer selfAddJob(){
		return ++jobs;
	}
	public void initJobs(){
		jobs = 0;
	}
	public Integer selfDelJob(){
		return --jobs;
	}
	public Integer getJobs(){
		return jobs;
	}
	@Override
	public void setFinished() {
		try {
			this.conn.close();
		} catch (SQLException e) {
			throw new IllegalStateException(e);
		}
//		SQLBatchOperation.currentBatch--;
		synchronized (sycLock) {
			sycLock.notifyAll();
		}
	}
}
